package com.niit.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object Spark_Stream_Winodw1 {

  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStream")
    val ssc = new StreamingContext(sparkConf,Seconds(3))
    ssc.sparkContext.setLogLevel("ERROR")

    val lines = ssc.socketTextStream("localhost",9999)
    val wordOne = lines.map((_,1))
  /*
    窗口时长和滑块步长都必须是采集周期的整数倍
   */
                            //  窗口时长      滑块步长
    val winDs = wordOne.window(Seconds(6), Seconds(6))

    val wordCount =  winDs.reduceByKey(_ + _)

    wordCount.print()

    ssc.start()
    ssc.awaitTermination()
  }

}
